package com.kakarote.hrm.task;

import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.TrackHits;
import com.alibaba.fastjson.JSONObject;
import com.kakarote.common.field.utils.EsUtil;
import com.kakarote.hrm.constant.HrmElasticSearchIndexEnum;
import com.kakarote.hrm.entity.VO.*;
import com.kakarote.hrm.mapper.HrmEmployeeMapper;
import com.kakarote.hrm.service.IHrmEmployeeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author 92487
 * @Date 2024/3/13 20:31
 */

@Component
@EnableScheduling
@EnableAsync
@Slf4j
public class ElasticSearchDataTask {

    @Resource
    private IHrmEmployeeService employeeService;
    @Autowired
    private HrmEmployeeMapper employeeMapper;

    @Scheduled(cron = "0 0 * * * ? ")
    public void elasticSearchDataSyncTask() throws IOException {
        Query query = new Query.Builder().matchAll(f -> f).build();
        SearchRequest request = new SearchRequest.Builder()
                .trackTotalHits(new TrackHits.Builder().enabled(true).build())
                .query(query)
                .source(s -> s.fetch(false))
                .size(0)
                .build();

        final SearchResponse<JSONObject> response = EsUtil.getClient().search(request, JSONObject.class);
        //总条数
        int totalSize = (int) response.hits().total().value();
        //每次查询条数
        int batchSize = 10000;
        int batchCount = totalSize % batchSize > 0 ? totalSize / batchSize + 1 : totalSize / batchSize;
        Set<Long> esEmpIds = new HashSet<>();
        for (int i = 0; i < batchCount; i++) {
            SearchRequest searchRequest = new SearchRequest.Builder()
                    .trackTotalHits(new TrackHits.Builder().enabled(true).build())
                    .query(query)
                    .source(s -> s.fetch(false))
                    .size(batchSize)
                    .from(i * batchSize)
                    .build();
            final SearchResponse<JSONObject> searchResponse = EsUtil.getClient().search(searchRequest, JSONObject.class);
            searchResponse.hits().hits().forEach(hit -> {
                esEmpIds.add(Long.parseLong(hit.id()));
            });
        }
        //数据库人员信息
        List<EmployeeElasticSearchVo> empList = employeeMapper.queryEmployeeElasticSearchData(new ArrayList<>());
        Set<Long> empIds;
        if (empList != null && !empList.isEmpty()) {
            empIds = empList.stream().map(EmployeeElasticSearchVo::getEmployeeId).collect(Collectors.toSet());
        } else {
            empIds = new HashSet<>();
        }
        List<Long> deleteIds = esEmpIds.stream().filter(id -> !empIds.contains(id)).collect(Collectors.toList());
        if (!deleteIds.isEmpty()) {
            EsUtil.deleteData(deleteIds, HrmElasticSearchIndexEnum.EMPLOYEE);
        }
        //初始化索引数据
        employeeService.initEmployeeEsData(new ArrayList<>());
        log.info("定时任务ES员工数据同步完成");
    }

}
